package com.dtruth.dataflow.bolt;

import java.util.Map;
import java.util.Set;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dtruth.dataflow.dao.RedisDAO;
import com.dtruth.dataflow.util.Env;

public class DataStatisticBolt extends BaseBasicBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	RedisDAO rd = null;

	/**
	 * On create
	 */
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		String host = Env.getInstance().getProperty("redis.thumbnail.host")
				.trim();
		int port = Integer.valueOf(Env.getInstance()
				.getProperty("redis.thumbnail.port").trim());
		int insNum = Integer.valueOf(Env.getInstance()
				.getProperty("redis.thumbnail.db").trim());
		rd = new RedisDAO(host, port, insNum);
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		@SuppressWarnings("unchecked")
		Map<String, Object> map = (Map<String, Object>) input.getValue(0);
		Object obj = map.get("thumbnail");
		String domain = map.get("domain").toString();
		if (obj != null) {
			JSONArray jarry = JSONObject.parseArray(obj.toString());
			for (int i = 0; i < jarry.size(); i++) {
				rd.zincrby("thumbnail:statistic", 1,
						domain + ":" + jarry.getString(i), 60 * 60 * 24 * 3);

			}
		}
		Set<String> members = rd.zrangeByScore("thumbnail:statistic", 3,
				Double.MAX_VALUE);
		for (String member : members) {
			String _member = member.substring(member.indexOf(":") + 1);
			rd.sadd("thumbnail:blacklist", _member);
			rd.zrem("thumbnail:statistic", member);
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}
}